package com.boat.starter.pulsar.config;

import org.apache.pulsar.client.api.AuthenticationFactory;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.core.env.Environment;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.converter.MappingJackson2MessageConverter;
import org.springframework.messaging.converter.MessageConverter;

import com.boat.starter.pulsar.constant.MQConstant;
import com.boat.starter.pulsar.core.PulsarTemplate;
import com.boat.starter.pulsar.properties.MQProperties;
import com.boat.starter.pulsar.properties.PulsarProperties;
import com.boat.starter.pulsar.support.JSON2Schema;
import com.boat.starter.pulsar.support.JdkUtils;

/**
 * Description: Pulsar自动配置
 *
 * @author: zuomin (myleszelic@outlook.com)
 * @date: 2021/07/16-17:06
 */
@Configuration
@ConditionalOnClass({PulsarClient.class})
@ConditionalOnProperty(prefix = MQConstant.PREFIX, value = "type", havingValue = MQConstant.MqType.PULSAR)
@Import({ListenerContainerConfiguration.class, TxListenerContainerConfiguration.class})
@EnableConfigurationProperties(value = {MQProperties.class})
public class PulsarAutoConfiguration implements ApplicationContextAware, InitializingBean {

    private static Logger log = LoggerFactory.getLogger(PulsarAutoConfiguration.class);

    public static final String PULSAR_CLIENT_BEAN_NAME = "defaultPulsarClient";
    public static final String PULSAR_TEMPLATE_BEAN_NAME = "pulsarTemplate";
    public static final String TX_PULSAR_CLIENT_BEAN_NAME = "txDefaultPulsarClient";
    public static final String TX_PULSAR_TEMPLATE_BEAN_NAME = "txPulsarTemplate";

    @Autowired
    private Environment environment;

    private ApplicationContext applicationContext;

    private MessageConverter messageConverter;

    @Bean(PULSAR_CLIENT_BEAN_NAME)
    @ConditionalOnProperty(prefix = MQConstant.PREFIX, value = {"pulsar.serviceUrl", "pulsar.authenticationToken"})
    public PulsarClient client(MQProperties mqProperties) {
        PulsarProperties pulsarProperties = mqProperties.getPulsar();
        try {
            return PulsarClient.builder()
                    .serviceUrl(pulsarProperties.getServiceUrl())
                    .authentication(AuthenticationFactory.token(pulsarProperties.getAuthenticationToken()))
                    .enableTlsHostnameVerification(pulsarProperties.isEnableTlsHostnameVerification())
                    .allowTlsInsecureConnection(pulsarProperties.isAllowTlsInsecureConnection())
                    .build();
        } catch (PulsarClientException e) {
            e.printStackTrace();
            throw new MessagingException("PulsarClient start fail, " + e.getMessage(), e);
        }
    }

    @Bean(TX_PULSAR_CLIENT_BEAN_NAME)
    @ConditionalOnProperty(prefix = MQConstant.PREFIX, value = {"txPulsar.serviceUrl", "txPulsar.authenticationToken"})
    public PulsarClient txClient(MQProperties mqProperties) {
        PulsarProperties pulsarProperties = mqProperties.getTxPulsar();
        try {
            return PulsarClient.builder()
                    .serviceUrl(pulsarProperties.getServiceUrl())
                    .authentication(AuthenticationFactory.token(pulsarProperties.getAuthenticationToken()))
                    .enableTlsHostnameVerification(pulsarProperties.isEnableTlsHostnameVerification())
                    .allowTlsInsecureConnection(pulsarProperties.isAllowTlsInsecureConnection())
                    .build();
        } catch (PulsarClientException e) {
            e.printStackTrace();
            throw new MessagingException("PulsarClient start fail, " + e.getMessage(), e);
        }
    }

    @Bean(name = PULSAR_TEMPLATE_BEAN_NAME, destroyMethod = "destroy")
    @ConditionalOnProperty(prefix = MQConstant.PREFIX, value = {"pulsar.serviceUrl", "pulsar.authenticationToken"})
    public PulsarTemplate pulsarTemplate(@Qualifier(PULSAR_CLIENT_BEAN_NAME) PulsarClient client, MQProperties mqProperties) {
        PulsarTemplate pulsarTemplate = new PulsarTemplate(client, mqProperties.getPulsar());
        pulsarTemplate.setEnvironment(environment);
        pulsarTemplate.initMessageConvert(messageConverter);
        return pulsarTemplate;
    }

    @Bean(name = TX_PULSAR_TEMPLATE_BEAN_NAME, destroyMethod = "destroy")
    @ConditionalOnProperty(prefix = MQConstant.PREFIX, value = {"txPulsar.serviceUrl", "txPulsar.authenticationToken"})
    public PulsarTemplate txPulsarTemplate(@Qualifier(TX_PULSAR_CLIENT_BEAN_NAME) PulsarClient txClient, MQProperties mqProperties) {
        PulsarTemplate pulsarTemplate = new PulsarTemplate(txClient, mqProperties.getTxPulsar());
        pulsarTemplate.setEnvironment(environment);
        pulsarTemplate.initMessageConvert(messageConverter);
        return pulsarTemplate;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
    	//取消jdk11警告
        JdkUtils.disableAccessWarnings();
        this.applicationContext = applicationContext;
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        MappingJackson2MessageConverter jackson2MessageConverter = new MappingJackson2MessageConverter();
        jackson2MessageConverter.setObjectMapper(JSON2Schema.getObjectMapper());
        jackson2MessageConverter.setSerializedPayloadClass(String.class);
        messageConverter = jackson2MessageConverter;
        
        log.info("boat-starter-pulsar加载成功");
    }
}
